package nio.reactor;

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.io.IOException;
import java.nio.ByteBuffer;
import java.nio.channels.SelectionKey;
import java.nio.channels.Selector;
import java.nio.channels.SocketChannel;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;

/**
 * @Desc:   多线程Handler处理器的实践案例
 * @author: cww
 * @DateTime: 2020-02-08 18:22
 */

public class MultiThreadEchoHandler implements Runnable{

    private static final Logger logger = LoggerFactory.getLogger(MultiThreadEchoHandler.class);

    final SocketChannel channel;
    final SelectionKey sk;
    final ByteBuffer byteBuffer = ByteBuffer.allocate(1024);
    static final int RECIEVING = 0, SENDING = 1;
    int state = RECIEVING;
    // 引入线程池
    static ExecutorService pool = Executors.newFixedThreadPool(1);

    public MultiThreadEchoHandler(Selector selector, SocketChannel channel) throws Exception {
        logger.info("currentThread: " + Thread.currentThread() + " ,初始化多线程 Handler ----");
        this.channel = channel;
        channel.configureBlocking(false);
        // 取得选择键，再设置感兴趣的 IO 事件
        this.sk = channel.register(selector, 0);
        // 初始化连接之后就把当前的 MultiThreadEchoHandler 注册进去，后面客户端的交互就由
        // 当前这个 Handler 进行专门处理收发事件了
        sk.attach(this);
        // 向 sk 选择键注册 read 就绪事件
        sk.interestOps(SelectionKey.OP_READ);
        // 使尚未返回的第一个选择操作立即返回
        selector.wakeup();
    }

    @Override
    public void run() {
        // 异步任务，在独立的线程池中执行 handler
        logger.info("currentThread: " + Thread.currentThread() + " ,提交异步任务到线程池----");
        pool.execute(new AsyncTask());
    }

    /**
     * 业务处理，不在反应器线程中执行
     */
    public synchronized void asyncRun() throws IOException {
        try {
            if (state == SENDING) {
                logger.info("currentThread: " + Thread.currentThread() + " ,发送模式----");
                // 写入通道
                channel.write(byteBuffer);
                // 写完后，准备开始从通道读，byteBuffer 切换成写入模式
                byteBuffer.clear();
                // 写完后，注册 read 就绪事件
                sk.interestOps(SelectionKey.OP_READ);
                // 写完之后，进入接收的状态
                state = RECIEVING;
            } else if (state == RECIEVING) {
                logger.info("currentThread: " + Thread.currentThread() + " ,接收模式----");
                int length = 0;
                while ((length = channel.read(byteBuffer)) > 0) {
                    logger.info(new String(byteBuffer.array(), 0, length));
                }
                // 读完之后，准备开始写入通道，byteBuffer 切换成读取模式
                byteBuffer.flip();
                // 读完之后，注册 write 就绪事件
                sk.interestOps(SelectionKey.OP_WRITE);
                // 读完后，进入发送状态
                state = SENDING;
            }
            // 处理结束了，这里不能关闭 select key，需要重复使用
//            sk.cancel();
        } catch (Exception e) {
            e.printStackTrace();
            // 这里如果不做判断进行关闭，则会一直循环抛错
            if(channel != null){
                logger.info("关闭通道");
                channel.close();
            }
        }
    }

    /**
     * 异步任务的内部类
     */
    class AsyncTask implements Runnable {

        @Override
        public void run() {
            logger.info("currentThread: " + Thread.currentThread() + " ,Handler 开始执行异步任务----");
            try {
                MultiThreadEchoHandler.this.asyncRun();
            } catch (IOException e) {
                e.printStackTrace();
            }
        }
    }
}
